package com.bsj.travel.config.rocketmq.producer;

import cn.hutool.core.text.CharPool;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import com.bsj.travel.cached.redis.RedisCached;
import com.bsj.travel.constant.BaseServerTopicConstants;
import com.bsj.travel.constants.RedisConstant;
import com.bsj.travel.def.common.DTO.mqtt.MqttEventPublish;
import com.bsj.travel.def.monitor.CloudLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.LongAdder;

/**
 * @author yinhao
 * @version 1.0
 * @description RocketMq消息发送
 * @date 2023/11/21
 */
@Slf4j
@Component
public class BaseMqPushUtil {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    @Autowired
    private RedisCached redisCached;

    /**
     * 发送成功的消息数量
     */
    private LongAdder msgSerNum = new LongAdder();


    /**
     * 发送设备下行消息
     * @param mqttEventPublish
     */
    public void sendMqttMsgDownTag(MqttEventPublish mqttEventPublish) {
        String nodeName = redisCached.get(RedisConstant.CONNECT_NODE + mqttEventPublish.getClient(), String.class);
        if (nodeName != null) {
            sendTopicBytes(BaseServerTopicConstants.TRAVEL_MQTT_MSG_TOPIC,
                    BaseServerTopicConstants.TRAVEL_MQTT_MSG_DOWN_TAG + "_" + nodeName,
                    JSON.toJSONBytes(mqttEventPublish));
        } else {
            // 转交给日志处理器
            CloudLog cloudLog = new CloudLog();
            cloudLog.setTs(System.currentTimeMillis());
            // 生成随机30位字符串当做事务ID
            cloudLog.setTraceId(RandomUtil.randomString(30));
            cloudLog.setServeType(1);
            cloudLog.setOperate(mqttEventPublish.getTopic());
            cloudLog.setIdentify(mqttEventPublish.getClient());
            Object data = mqttEventPublish.getData();
            String content;
            if (data instanceof String) {
                content = data.toString();
            } else {
                content = JSON.toJSONString(data);
            }
            cloudLog.setContent(content);
            // 离线下发失败
            cloudLog.setCode(9201);
            sendTopicBytes(BaseServerTopicConstants.TRAVEL_MQTT_LOG_MSG_RECORD_TOPIC,
                    BaseServerTopicConstants.TRAVEL_MQTT_LOG_MSG_RECORD_TAG, com.alibaba.fastjson2.JSON.toJSONBytes(cloudLog));
        }
    }


    /**
     * 根据主题 tag 发送消息
     *
     * @param bytes
     * @param tag
     */
    public void sendTopicBytes(String topic, String tag, byte[] bytes) {
        //消费者的大tag和小tag和数据
        Message msg = new Message(topic, tag, bytes);
        try {
            send(msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 内部真正的发送
     *
     * @param msg
     * @throws Exception
     */
    private void send(Message msg) throws Exception {
        SendResult result = defaultMQProducer.send(msg);
        if (!SendStatus.SEND_OK.equals(result.getSendStatus())) {
            throw new Exception("MQ发送失败");
        }
        String messageId = result.getMsgId();
        msgSerNum.increment();
        log.info("消息发送成功数量：{},消息ID：{},消息Topic： {}, 消息tag： {}, 发送的值：{}", msgSerNum.sum(), messageId,
                msg.getTopic(), msg.getTags(), new String(msg.getBody()));
    }
}
